package com.corn.flink.lesson6;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * @author JimWu
 * @date 2023/4/17 15:45
 **/
public class ConnectOperatorDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Integer> ds1 = env.fromElements(1, 2, 3);
        DataStreamSource<Long> ds2 = env.fromElements(1L, 2L, 3L);

        // 调用connect 数据流链接 返回ConnectedStreams
        ConnectedStreams<Integer, Long> connectStream = ds1.connect(ds2);

        // ConnectedStreams.map方法传入CoMapFunction 分别对不同的连接流进行不同的数据结构处理,再统一数据结构输出到一个DataStream中
        SingleOutputStreamOperator<String> resultDs = connectStream.map(new CoMapFunction<Integer, Long, String>() {
            @Override
            public String map1(Integer value) throws Exception {
                return "Int: " + value;
            }

            @Override
            public String map2(Long value) throws Exception {
                return "Long: " + value;
            }
        });
        resultDs.print("resultStream");
        env.execute();
    }
}
